package com.uh.rds.tester.standalone;

import com.uh.rds.testing.logger.TestLoggerFactory;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import redis.clients.jedis.Connection;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RedisClusterExample {

    final static int TOTAL_THREADS = 3;
    final static int KEYS_PER_THREAD = 100000;
    final static int RETRY_COUNT = 3;

    // Redis集群节点
    private static HostAndPort CLUSTER_NODE;

    // 初始化集群节点
    static {
        CLUSTER_NODE = new HostAndPort("127.0.0.1", 6331);
    }

    // 连接池配置
    private static GenericObjectPoolConfig<Connection> getPoolConfig() {
        GenericObjectPoolConfig<Connection> poolConfig = new GenericObjectPoolConfig<Connection>();
        // 最大连接数
        poolConfig.setMaxTotal(100);
        // 最大空闲连接数
        poolConfig.setMaxIdle(20);
        // 最小空闲连接数
        poolConfig.setMinIdle(5);
        // 开启空闲资源监测
        poolConfig.setTestWhileIdle(true);
        return poolConfig;
    }

    public static void main(String[] args) {
        JedisCluster jedisCluster = null;
        try {
            // 创建JedisCluster实例，内部包含连接池
            jedisCluster = new JedisCluster(CLUSTER_NODE, 5000, 5000, 3, "123456",
                     getPoolConfig());


            // 创建并启动线程
            ClusterSetThread[] threads = new ClusterSetThread[TOTAL_THREADS];
            for (int i = 0; i < TOTAL_THREADS; i++) {
                threads[i] = new ClusterSetThread(jedisCluster, i, KEYS_PER_THREAD);
                threads[i].start();
            }
            System.out.println("\n~~~~~ All ClusterSetThread started.");
            // 等待所有线程结束
            for (int i = 0; i < TOTAL_THREADS; i++) {
                try {
                    threads[i].join();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
            System.out.println("\n~~~~~ All ClusterSetThread completed.");

            // 校验所有key
            ClusterGetValidateThread[] validateThreads = new ClusterGetValidateThread[TOTAL_THREADS];
            for (int i = 0; i < TOTAL_THREADS; i++) {
                validateThreads[i] = new ClusterGetValidateThread(jedisCluster, i, KEYS_PER_THREAD);
                validateThreads[i].start();
            }
            System.out.println("\n~~~~~ All ClusterGetValidateThread started.");
            for (int i = 0; i < TOTAL_THREADS; i++) {
                try {
                    validateThreads[i].join();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
            System.out.println("\n~~~~~ All ClusterGetValidateThread completed.");

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭连接
            if (jedisCluster != null) {
                try {
                    jedisCluster.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }


    static class ClusterSetThread extends Thread {
        private final Logger logger = TestLoggerFactory.getLogger(ClusterSetThread.class);
        private final JedisCluster jedisCluster;
        private final int threadNum;
        private final int totalKeys;

        public ClusterSetThread(JedisCluster jedisCluster, int threadNum, int totalKeys) {
            this.jedisCluster = jedisCluster;
            this.threadNum = threadNum;
            this.totalKeys = totalKeys;
        }

        @Override
        public void run() {
            //System.out.println("ClusterSetThread " + threadNum + " started.");
            for (int i = 0; i < totalKeys; i++) {
                String key = "Thread" + threadNum + "_key" + i;
                String value = "value" + i + "_from_thread" + threadNum;
                retrySet(key, value, RETRY_COUNT);
                // 每运行200个key，打印一次进度
                if (threadNum == 0 && (i + 1) % (10000) == 0) {
                    System.out.print("*");
                }
            }
        }

        private void retrySet(String key, String value, int retries) {
            for (int i = 0; i < retries; i++) {
                try {
                    String result = jedisCluster.set(key, value);
                    if ("OK".equals(result)) {
                        return;
                    }
                    else {
                        logger.error("Failed to set key: {}, return {}", key, result);
                    }
                } catch (Exception e) {
                    logger.error("Error setting key: {}, attempt {}", key, i + 1, e);
                }
                try {
                    Thread.sleep(100); // Wait before retrying
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                }
            }
            logger.error("Failed to set key: {} after {} retries.", key, retries);
        }
    }


    public static class ClusterGetValidateThread extends Thread {
        private final Logger logger = TestLoggerFactory.getLogger(ClusterGetValidateThread.class);
        private final JedisCluster jedisCluster;
        private final int threadNum;
        private final int totalKeys;

        public ClusterGetValidateThread(JedisCluster jedisCluster, int threadNum, int totalKeys) {
            this.jedisCluster = jedisCluster;
            this.threadNum = threadNum;
            this.totalKeys = totalKeys;
        }

        @Override
        public void run() {
            //System.out.println("ClusterGetValidateThread " + threadNum + " started.");
            for (int i = 0; i < totalKeys; i++) {
                String key = "Thread" + threadNum + "_key" + i;
                String expectedValue = "value" + i + "_from_thread" + threadNum;
                String actualValue = null;
                try {
                    actualValue = jedisCluster.get(key);
                } catch (Exception e) {
                    logger.error("Error getting key: {}", key, e);
                }
                if (actualValue == null) {
                    logger.error("Key not found: {}", key);
                } else if (!expectedValue.equals(actualValue)) {
                    logger.error("Value mismatch for key: {}. Expected: {}, Actual: {}", key, expectedValue, actualValue);
                }

                // 每运行200个key，打印一次进度
                if(threadNum == 0 && (i + 1) % (10000) == 0) {
                    System.out.print("#");
                }
            }
        }
    }

}
